package com.ruge.kafka.consume;

import com.ruge.tool.date.DateTool;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class KafkaConsumer {
    int i = 0;
    @Resource
    private JdbcTemplate jdbcTemplate;
    @Resource
    private ScheduledExecutorService scheduledExecutorService;
    // 消费监听
    @KafkaListener(topics = {"CDP_USER_PRFOFILE_2_CMP"}, containerFactory = "kafkaListenerContainerFactory2")
    public void onMessage(List<ConsumerRecord<String, String>> records, Acknowledgment ack) throws IOException {
        CountDownLatch countDownLatch = new CountDownLatch(records.size());
        System.out.println(DateTool.getDateTimeInstance());
        FileUtils.writeStringToFile(new File("D:\\QMDownload\\demo.txt"),DateTool.getDateTimeInstance()+ "\r\n",true);
//        System.out.println("此线程消费" + records.size() + "条消息----线程名:" + Thread.currentThread().getName());
//        records.forEach(record -> System.out.println("topic名称:"+record.topic()+"\n"+"分区位置:"+record.partition()+"\n"+"key:"+record.key()+"\n"+"偏移量:"+record.offset()+"\n"+"消息内容:"+record.value()));
        records.forEach(record -> {
            submitConsumeTask(record, countDownLatch);
        });
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            log.error("countDownLatch exception ", e);
        }
        ack.acknowledge();
    }

    private void submitConsumeTask(ConsumerRecord<String, String> record, CountDownLatch countDownLatch) {
        scheduledExecutorService.schedule(() -> {
                    try {
//                        FileUtils.writeStringToFile(new File("D:\\QMDownload\\demo.txt"),record.offset()+ "\r\n",true);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }finally {
                        countDownLatch.countDown();
                    }
                }
                , 3, TimeUnit.SECONDS);
    }
}